package com.atguigu.gmall.realtime.app

import java.util

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.gmall.realtime.util.{MyKafkaSink, MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

object BaseDbApp {
  //1  环境
  //2  读redis的偏移量
  //3  根据偏移加载kafka流
  //4  把偏移量结束点提取出来
  //5  把kafka转换为结构化对象流
  // 6  根据数据类型（维度事实） 保存到不同位置中
  // 7 提交偏移量
  def main(args: Array[String]): Unit = {
      //1  环境
      val sparkConf: SparkConf = new SparkConf().setAppName("base_db_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    val topic = "ODS_BASE_DB_C"
    val groupId = "base_db_group"

    //2  读redis的偏移量
    val offsetMapFromRedis: Map[TopicPartition, Long] = OffsetManager.getOffset(topic,groupId)

      //3  根据偏移加载kafka流
      var inputDstream: InputDStream[ConsumerRecord[String, String]]=null
     if(offsetMapFromRedis!=null&&offsetMapFromRedis.size>0){
       inputDstream= MyKafkaUtil.getKafkaStream(topic,ssc,offsetMapFromRedis,groupId)
     }else{
       inputDstream= MyKafkaUtil.getKafkaStream(topic,ssc,groupId)
     }
      //4  把偏移量结束点提取出来
      var offsetRanges: Array[OffsetRange]=null
    val inputDstreamAfterOffset: DStream[ConsumerRecord[String, String]] = inputDstream.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }



       //5  把kafka转换为结构化对象流( 通用，专用)
    val jsonObjDstream: DStream[JSONObject] = inputDstreamAfterOffset.map { record =>
      val jsonString: String = record.value()
      val jsonObj: JSONObject = JSON.parseObject(jsonString)
      jsonObj
    }
  //  jsonObjDstream.print(1000)
       // 6  根据数据类型（维度事实） 保存到不同位置中

   // X    driver   启动只执行一次    // 如果写在此处 数据变化 而程序中的变量是不会变得
   //  val jedisclient: Jedis = RedisUtil.getJedisClient
   //    val dimTables = jedisclient.smembers("dim_tables")

    jsonObjDstream.foreachRDD{rdd=>
      //A     ?    driver 周期性执行
      val jedisclient: Jedis = RedisUtil.getJedisClient
      val dimTablesFromRedis = jedisclient.smembers("dim_tables")  //广播变量方式
   //   println(s"维度表清单：$dimTablesFromRedis")
      val dimTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTablesFromRedis)

      rdd.foreachPartition{jsonObjItr=>
        val dimTables: util.Set[String] = dimTablesBC.value
        //B   ?     executor
   //   val jedisclient: Jedis = RedisUtil.getJedisClient
    //    val dimTables = jedisclient.smembers("dim_tables")

        val jedis: Jedis = RedisUtil.getJedisClient
        val jsonObjList: List[JSONObject] = jsonObjItr.toList

        for ( jsonObj<- jsonObjList ) {
          //C
       //   val jedisclient: Jedis = RedisUtil.getJedisClient
       //   val dimTables = jedisclient.smembers("dim_tables")


          val optType: String = jsonObj.getString("type")

          if(Array("INSERT","UPDATE").contains(optType)){
            val dataTableName: String = jsonObj.getString("table")
            if(dimTables.contains(dataTableName)){
              //维度数据  ==> 发送redis
              // redis type? string   key ?  user_info:[101]  value?  json  field/score? 无
              // 写？  set  读取？ get    过期时间？ 不过期
              val jSONArray: JSONArray = jsonObj.getJSONArray("pkNames") //提取主键名
              val pkName: String = jSONArray.getString(0)  //假设所有的表没有复合主键
              val dataJsonArr: JSONArray = jsonObj.getJSONArray("data")  //提取数据 （可能一个大json中包含多条变更数据)
              for( i <- 0 to dataJsonArr.size()-1  ){
                val dataJsonObj: JSONObject = dataJsonArr.getJSONObject(i)
                val pkValue: String = dataJsonObj.getString(pkName) // 根据主键名提取主键值
                val key = dataTableName+":"+pkValue
                jedis.set(key,dataJsonObj.toJSONString)
              }
            }else{
              //事实数据  ==> 发送kafka
              // 按主题进行切分 ：   层名_表名_操作标识(I / U / D)
              println(jsonObj)
              var optFlag:String =null
              if("INSERT"==optType){
                optFlag="I"
              }else if ("UPDATE"==optType){
                optFlag="U"
              }
              val jSONArray: JSONArray = jsonObj.getJSONArray("pkNames") //提取主键名
              val pkName: String = jSONArray.getString(0)  //假设所有的表没有复合主键
              val topic=s"DWD_${dataTableName.toUpperCase}_$optFlag"
              val dataJsonArr: JSONArray = jsonObj.getJSONArray("data")  //提取数据 （可能一个大json中包含多条变更数据)
              for( i <- 0 to dataJsonArr.size()-1  ){
                val dataJsonObj: JSONObject = dataJsonArr.getJSONObject(i)
                val pkValue: String = dataJsonObj.getString(pkName) // 根据主键名提取主键值
                val key = dataTableName+":"+pkValue
                 MyKafkaSink.send(topic,key, dataJsonObj.toJSONString )
                 if(topic=="DWD_ORDER_INFO_I"||topic=="DWD_ORDER_DETAIL_I"){
                   Thread.sleep(150)
                 }
              }
            }
          }

        }

        jedis.close()
      }
      //提价偏移量
      OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }


      // 7 提交偏移量

    //8 启动环境
    ssc.start()
    ssc.awaitTermination()


  }

}
